package com.cqhfan.imservice.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.cqhfan.api.clients.user.UserClient;
import com.cqhfan.core.constans.DateConstants;
import com.cqhfan.core.constans.MqConstants;
import com.cqhfan.core.domain.dto.PageDTO;
import com.cqhfan.core.enums.ChatTypeEnum;
import com.cqhfan.core.enums.MessageTypeEnum;
import com.cqhfan.core.exceptions.BizException;
import com.cqhfan.core.utils.UserContext;
import com.cqhfan.imservice.handler.UserConnectionSession;
import com.cqhfan.imservice.mapper.ImMessageMapper;
import com.cqhfan.imservice.service.ImMessageService;
import com.cqhfan.mq.strategy.MqMessageHelper;
import com.cqhfan.pojo.im.dto.TogetherWatchMessageDTO;
import com.cqhfan.pojo.im.dto.WebSocketMessageDTO;
import com.cqhfan.pojo.im.po.ImMessage;
import com.cqhfan.pojo.im.query.ChatHistoryQuery;
import com.cqhfan.pojo.im.vo.HistoryMessageVO;
import com.cqhfan.pojo.user.po.User;
import com.cqhfan.redis.constants.RedisConstants;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author 眭传洪
 * @create 2025-08-16 2:17
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class ImMessageServiceImpl extends ServiceImpl<ImMessageMapper, ImMessage> implements ImMessageService {

    private final MqMessageHelper mqMessageHelper;

    private final StringRedisTemplate stringRedisTemplate;

    private final UserClient userClient;

    @Override
    @Transactional
    public void sendMessage(ImMessage body) {
        // 保存消息
        LocalDateTime now = LocalDateTime.now();
        String chatTime = DateUtil.format(now, DateConstants.DATE_TIME_FORMAT);
        Long userId = UserContext.getUser();
        body.setSenderId(userId);
        body.setCreateTime(now);
        body.setChatTime(chatTime);
        Long receiverId = body.getReceiverId();
        // 设置是否已读
        String key = RedisConstants.IM.UPDATE_CURRENT_CHAT_LIST_KEY;
        Object obj = stringRedisTemplate.opsForHash().get(key, receiverId.toString());
        Long chatListId = null;
        if (obj != null) {
            chatListId = Long.valueOf(obj.toString());
        }
        if (body.getChatType() == ChatTypeEnum.SINGLE) {
            // 单聊标记是否已读
            body.setIfRead(chatListId != null && NumberUtil.equals(chatListId, body.getChatListId()));
        } else {
            // 群聊直接累加未读数量
            if (chatListId == null || !NumberUtil.equals(chatListId, body.getChatListId())) {
                key = RedisConstants.IM.GROUP_MESSAGE_UNREAD_QUANTITY_KEY_PREFIX + body.getGroupId();
                stringRedisTemplate.opsForHash().increment(key, receiverId.toString(), 1);
            }
        }
        save(body);

        // 构建即时消息发送mq
        List<User> users = userClient.listByIds(Collections.singletonList(userId));
        User user = users.get(0);
        WebSocketMessageDTO messageDTO = WebSocketMessageDTO.builder()
                .chatType(body.getChatType().getValue())
                .messageType(body.getMessageType().getValue())
                .groupId(body.getGroupId())
                .content(body.getContent())
                .senderUserNickName(user.getNickName())
                .senderUserAvatar(user.getAvatar())
                .senderId(body.getSenderId())
                .receiverId(receiverId)
                .chatTime(chatTime)
                .chatListId(body.getChatListId())
                .build();
        mqMessageHelper.sendMessage(
                MqConstants.EXCHANGE.CHAT_SEND_EXCHANGE,
                null,
                messageDTO
        );
    }

    @Override
    public void listenSendMsg(String msg) {
        WebSocketMessageDTO message = JSONUtil.toBean(msg, WebSocketMessageDTO.class);
        // 根据单聊和群聊情况发送消息
        ChatTypeEnum chatType = ChatTypeEnum.fromValue(message.getChatType());
        if (ChatTypeEnum.SINGLE == chatType) {
            Long receiverId = message.getReceiverId();
            WebSocketSession receiveSession = UserConnectionSession.getConnectionRelation(receiverId);
            if (receiveSession != null && receiveSession.isOpen()) {
                try {
                    receiveSession.sendMessage(new TextMessage(JSONUtil.toJsonStr(message)));
                } catch (IOException e) {
                    throw new BizException(e.getMessage());
                }
            }
        } else {
            // 群聊消息发送
            sendGroupMessage(msg, message.getGroupId(), message.getSenderId());
        }
    }

    @Override
    public void sendGroupMessage(String msg, Long groupId, Long senderId) {
        // 查询redis中群的人
        String key = RedisConstants.IM.GROUP_MEMBERS_KEY_PREFIX + groupId;
        Set<String> cacheGroupUserIds = stringRedisTemplate.opsForSet().members(key);
        if (CollUtil.isEmpty(cacheGroupUserIds)) {
            return;
        }
        Set<Long> userIds = cacheGroupUserIds.stream().map(Long::valueOf).collect(Collectors.toSet());

        // 拿到本机下所有的人
        Set<Long> currentNodeUserIds = UserConnectionSession.getConnectionUserIds();

        // 求出交集并发送即时消息
        Collection<Long> currentNodeGroupUserIds = CollUtil.intersection(userIds, currentNodeUserIds);
        currentNodeGroupUserIds.remove(senderId);
        log.info("群id {} 下的本机用户{}", groupId, currentNodeGroupUserIds);

        // 发送消息
        for (Long userId : currentNodeGroupUserIds) {
            WebSocketSession session = UserConnectionSession.getConnectionRelation(userId);
            if (session == null || !session.isOpen()) {
                continue;
            }
            try {
                session.sendMessage(new TextMessage(msg));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public PageDTO<HistoryMessageVO> history(ChatHistoryQuery query) {
        Long chatListId = query.getChatListId();
        Integer pageNum = query.getPageNum();
        Integer pageSize = query.getPageSize();

        // 查询消息
        Page<ImMessage> page = lambdaQuery()
                .eq(ImMessage::getChatListId, chatListId)
                .orderByDesc(ImMessage::getCreateTime)
                .page(new Page<>(pageNum, pageSize));
        List<ImMessage> records = page.getRecords();
        if (CollUtil.isEmpty(records)) {
            return PageDTO.empty();
        }

        // 转换vo
        List<HistoryMessageVO> result = records.stream()
                .map(item -> BeanUtil.copyProperties(item, HistoryMessageVO.class))
                .toList();
        return PageDTO.of(page.getPages(), page.getTotal(), result);
    }

    @Override
    public void togetherWatchHandler(TogetherWatchMessageDTO togetherWatchMessage) {
        // 查询邀请方
        Long userId = UserContext.getUser();
        List<User> users = userClient.listByIds(Collections.singletonList(userId));
        User user = users.get(0);

        // 构建消息
        WebSocketMessageDTO message = WebSocketMessageDTO.builder()
                .senderId(userId)
                .receiverId(togetherWatchMessage.getReceiveId())
                .senderUserNickName(user.getNickName())
                .senderUserAvatar(user.getAvatar())
                .messageType(MessageTypeEnum.TOGETHER_WATCH.getValue())
                .chatType(ChatTypeEnum.SINGLE.getValue())
                .togetherWatchMessage(togetherWatchMessage)
                .build();

        // 发送消息
        mqMessageHelper.sendMessage(
                MqConstants.EXCHANGE.CHAT_SEND_EXCHANGE,
                null,
                message
        );
    }

}
